package com.atguigu.gmall.realtime.common.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.bean.DimJoinFunction;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.HBaseUtil;
import com.atguigu.gmall.realtime.common.util.RedisUtil;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @author Felix
 * @date 2024/7/01
 * 发送异步请求进行维度关联的模板类
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimJoinFunction<T> {

    private StatefulRedisConnection<String,String> redisAsyncConn;
    private AsyncConnection hbaseAsyncConn;

    @Override
    public void open(Configuration parameters) throws Exception {
        redisAsyncConn = RedisUtil.getAsyncRedisConnection();
        hbaseAsyncConn = HBaseUtil.getAsyncHBaseConnection();
    }

    @Override
    public void close() throws Exception {
        RedisUtil.closeAsyncRedisConnection(redisAsyncConn);
        HBaseUtil.closeAsyncHBaseConnection(hbaseAsyncConn);
    }

    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
        /*//根据流中对象获取要关联的维度的主键
        String key = getRowKey(obj);
        //发送异步请求到Redis中获取维度数据
        JSONObject dimJsonObj = RedisUtil.readDimAsync(redisAsyncConn, getTableName(), key);
        if(dimJsonObj != null){
            //如果在Redis中找到了要关联的维度，直接将其返回(缓存命中)
            System.out.println("~~~从Redis中获取"+getTableName()+"表的"+key+"维度~~~");
        }else{
            //如果在Redis中没有找到要关联的维度，发送异步请求到HBase中查询维度数据
            dimJsonObj = HBaseUtil.readDimAsync(hbaseAsyncConn, Constant.HBASE_NAMESPACE,getTableName(),key);
            if(dimJsonObj != null){
                //将查询出来的维度以异步的方式放到Redis中保存起来，方便下次查询使用
                System.out.println("~~~从HBase中获取"+getTableName()+"表的"+key+"维度~~~");
                RedisUtil.writeDimAsync(redisAsyncConn,getTableName(),key,dimJsonObj);
            }else {
                System.out.println("~~~没有找到"+getTableName()+"表的"+key+"维度~~~");
            }
        }
        //将维度属性补充到流中对象上
        if(dimJsonObj != null){
            addDims(obj,dimJsonObj);
        }
        //获取数据库交互的结果并发送给ResultFuture的回调函数  向下游传递数据
        resultFuture.complete(Collections.singleton(obj));*/

        //创建异步编排对象，执行线程任务  supply开头的方法有返回值
        CompletableFuture.supplyAsync(
                new Supplier<JSONObject>() {
                    @Override
                    public JSONObject get() {
                        //发送异步请求到Redis中获取维度数据
                        JSONObject dimJsonObj = RedisUtil.readDimAsync(redisAsyncConn, getTableName(), getRowKey(obj));
                        return dimJsonObj;
                    }
                }
        ).thenApplyAsync(
                //执行线程任务   有入参，有返回值   上一个线程任务的返回结果会作为当前线程任务的入参
                new Function<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject apply(JSONObject dimJsonObj) {
                        if(dimJsonObj != null){
                            //如果在Redis中找到了要关联的维度，直接将其返回(缓存命中)
                            System.out.println("~~~从Redis中获取"+getTableName()+"表的"+getRowKey(obj)+"维度~~~");
                        }else{
                            //如果在Redis中没有找到要关联的维度，发送异步请求到HBase中查询维度数据
                            dimJsonObj = HBaseUtil.readDimAsync(hbaseAsyncConn, Constant.HBASE_NAMESPACE,getTableName(),getRowKey(obj));
                            if(dimJsonObj != null){
                                //将查询出来的维度以异步的方式放到Redis中保存起来，方便下次查询使用
                                System.out.println("~~~从HBase中获取"+getTableName()+"表的"+getRowKey(obj)+"维度~~~");
                                RedisUtil.writeDimAsync(redisAsyncConn,getTableName(),getRowKey(obj),dimJsonObj);
                            }else {
                                System.out.println("~~~没有找到"+getTableName()+"表的"+getRowKey(obj)+"维度~~~");
                            }
                        }
                        return dimJsonObj;
                    }
                }
        ).thenAcceptAsync(
                //执行线程任务  有入参，无返回值
                new Consumer<JSONObject>() {
                    @Override
                    public void accept(JSONObject dimJsonObj) {
                        if(dimJsonObj != null){
                            addDims(obj,dimJsonObj);
                        }
                        //获取数据库交互的结果并发送给ResultFuture的回调函数  向下游传递数据
                        resultFuture.complete(Collections.singleton(obj));
                    }
                }
        );

    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        throw new RuntimeException("1.先确认并行度是否设置都正确\n" +
                                   "2.确认该启动的进程是否都启动了 zk、kafka、maxwell、hdfs、hbase、redis、dwd下单事实表处理应用、当前应用\n" +
                                   "3.找昌平彦祖");
    }
}
